package com.justgame.project.task;

import com.justgame.mq.client.service.RabbitMqSendService;
import com.justgame.project.cloud.base.entity.MgMessageQueueLog;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;

import java.util.Calendar;
import java.util.Date;
import java.util.List;

/**
 * 消息队列定时任务
 * @author ZhouWeiBo
 * @create 2022/6/1
 */
@Configuration
@Slf4j
@EnableScheduling
public class RabbitTask {

    @Autowired
    private MongoTemplate mongoTemplate;

    @Autowired
    private RabbitMqSendService rabbitMqSendService;

    /**
     * 重新投递消息到MQ broker
     * 频率：3m/次
     */
    @Async("taskExecutor")
    @Scheduled(fixedRate = 1000 * 60 * 3)
    protected void redispatchMessage(){
        log.info("正在重新投递消息到MQ broker端");
        /* 计算1分钟前的时间 */
        Calendar beforeCalendar = Calendar.getInstance();
        beforeCalendar.add(Calendar.MINUTE,-1);
        Date beforeDate = beforeCalendar.getTime();
        /* 查询1分钟前未投递消息成功的 */
        Query query = new Query();
        query.addCriteria(
                Criteria.where("dispatchSuccess").is(false).and("createTime").lte(beforeDate)
        );
        List<MgMessageQueueLog> records = mongoTemplate.find(query, MgMessageQueueLog.class);
        records.forEach( record -> {
            CorrelationData correlationData = new CorrelationData();
            correlationData.setId(record.getId());
            log.info("投递CorrelationDataId:{}",record.getId());
            rabbitMqSendService.resendMsg(record.getRoutingKey(),record.getMessageBody(),correlationData);
        });
        log.info("共{}条消息重新投递完成！",records.size());
    }
}
